package ruc.InfluxDB.read;

import okhttp3.Request;
import okhttp3.Response;
import ruc.Consts.constant;
import ruc.utils.IOUtils;
import ruc.InfluxDB.InfluxDB_Common;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

public class InfluxDB_ConcurrentRead {
    private int bound;
    private String tbName;
    private String queryURL = "";

    public InfluxDB_ConcurrentRead(String IP, String port, String dbName, String tbName, int bound) {
        String baseUrl = String.format("http://%s:%s", IP, port);
        this.queryURL = baseUrl + "/query?db=" + dbName;
        this.tbName = tbName;
        this.bound = bound;
    }

    public long convertAndSend(String eSql) throws IOException {
        String tmp = queryURL + "&q=" + eSql;

        Request request = new Request.Builder()
                .url(tmp)
                .build();
        long st_time = System.nanoTime();
        Response response = InfluxDB_Common.getClient().newCall(request).execute();
        response.close();

        return (System.nanoTime() - st_time) / constant.Ns2Ms;
    }

    public long query(int limit, int offset) throws IOException {
        String sqlFormat = "select * from %s limit %d offset %d";
        String eSql = String.format(sqlFormat, tbName, limit, offset);

        return convertAndSend(eSql);
    }

    public void concurrentRead(int client[]) throws InterruptedException {
        Random ran = new Random();

        for (int i = 0; i < client.length; i++) {
            CountDownLatch CDL = new CountDownLatch(client[i]);
            AtomicLong AL = new AtomicLong(0);
            for (int k = 0; k < client[i]; k++) {
                new Thread(() -> {
                    try {
                        AL.addAndGet(query(ran.nextInt(bound), ran.nextInt(bound)));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    CDL.countDown();
                }).start();
            }

            CDL.await();
            System.out.printf("[TSBench-D] [InfluxDB_ConcurrentRead] >>>>> ThreadNum:%d, average response time : %d >>>>> \n", client[i], AL.longValue() / (long) client[i]);
            IOUtils.WriteData(constant.GetWriteFilePath("InfluxDB", "concurrentRead"),
                    new String[]{String.valueOf(System.nanoTime()), String.valueOf(client[i]), String.valueOf(AL.longValue() / (long) client[i])});
        }
    }

    public static void main(String[] args) throws InterruptedException {
        InfluxDB_ConcurrentRead db = new InfluxDB_ConcurrentRead(constant.IP, constant.InfluxDB_Port, constant.dbName, constant.Concurrent_tbName, constant.bound);
        db.concurrentRead(constant.ConcurrentReadThreadNum);
    }
}
